package com.haozhen.streaming

/**
  * @author haozhen
  * @email haozh@ync1.com
  * @date 2021/1/16  22:57
  */
object SocketWindowDemo {

  def main(args: Array[String]): Unit = {
    import org.apache.log4j.{Level, Logger}
    import org.apache.spark.SparkConf
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    Logger.getLogger("org").setLevel(Level.WARN)


    val conf: SparkConf = new SparkConf().setAppName("SocketStream").setMaster("local[*]")
    val ssc = new StreamingContext(conf,Seconds(5))

    //linux 开启 socket nc -lk 9999
    val lines = ssc.socketTextStream("localhost",9999)
    lines.foreachRDD{
       (rdd,time)=>
        println(s"rdd = ${rdd.id}  time = $time")
        rdd.foreach(println)
    }
    println("------------")
    lines.reduceByWindow(_+" "+_,Seconds(10),Seconds(5)).print()
    println("------------")
    ssc.start()

    ssc.awaitTermination()
  }
}
